package com.hyt.it.ogt.opertaion.track.queue;

import com.hyt.it.ogt.opertaion.track.executor.OperationTrackExecutorThreadFactory;
import com.hyt.it.ogt.opertaion.track.mq.OperationTrackMqProducer;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

/**
 * @author 朱振宇
 * @since 2022-12-02
 */
@Slf4j
@Component
public class OperationTrackBlockQueue implements InitializingBean, ApplicationContextAware {

    private ApplicationContext applicationContext;

    @Resource
    private OperationTrackMqProducer operationTrackMqProducer;

    protected final static BlockingQueue<Object> logSaveQueue = new LinkedBlockingQueue<>(65536);

    private static final long WAIT_TIME = 50;

    private static final int BATCH_SIZE = 100;

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }

    @Override
    public void afterPropertiesSet() {
        if(applicationContext.getBeanProvider(RabbitTemplate.class).getIfAvailable() == null) {
            log.error("没有配置RabbitTemplate，OperationTrackBlockQueue初始化失败！");
            return;
        }
        new OperationTrackExecutorThreadFactory("OperationTrackLogThreadFactory").newThread(new LogPushTask()).start();
    }

    public static void producer(Object obj) {
        try {
            logSaveQueue.put(obj);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    @SuppressWarnings("InfiniteLoopStatement")
    final class LogPushTask implements Runnable {
        @Override
        public void run() {
            List<Object> msgList = null;
            while(true) {
                try {
                    List<Object> msg = (List<Object>) logSaveQueue.take();
                    msgList = new ArrayList<>(BATCH_SIZE);
                    msgList.addAll(msg);
                    while (msgList.size() < BATCH_SIZE && Objects.nonNull((msg = (List<Object>) logSaveQueue.poll(WAIT_TIME, TimeUnit.MILLISECONDS)))) {
                        msgList.addAll(msg);
                    }
                } catch (InterruptedException e) {
                    log.error("日志推送消费异常");
                    e.printStackTrace();
                }
                if(Objects.nonNull(msgList) && !msgList.isEmpty()) {
                    try {
                        operationTrackMqProducer.sendMessage(msgList);
                    } catch (Exception e) {
                        log.error("mq推送异常");
                        e.printStackTrace();
                    }
                }
            }
        }
    }

}
